#include "server.h"
#include "OperationNode.h"
#include "Column.h"
#include "Consumer.h"

Consumer::Consumer(OperationNode* operationsRoot, Server* server):
		_operationsRoot(operationsRoot),
		_server(server) {}

Consumer::~Consumer() {
	delete _operationsRoot;
}

template <typename T>
void consumeDataByServer(Server* server, int column_index, const Column& column);

template <>
void consumeDataByServer<int32_t>(Server* server, int column_index, const Column& column) {
	server->ConsumeInts(column_index, column.size(), column.data<int32_t>());
}

template <>
void consumeDataByServer<double>(Server* server, int column_index, const Column& column) {
	server->ConsumeDoubles(column_index, column.size(), column.data<double>());
}

template <>
void consumeDataByServer<bool>(Server* server, int column_index, const Column& column) {
	server->ConsumeByteBools(column_index, column.size(), column.data<bool>());
}

void Consumer::consume() {
	SharedArray<Column> in = _operationsRoot->pull();

	if (in.size() == 0)
		return;

	while (in[0].size() != 0) {
		for (int i = 0; i < in.size(); ++i) {
			switch (in[i].type()) {
				case INT:
					consumeDataByServer<int32_t>(_server, i, in[i]);
					break;

				case DOUBLE:
					consumeDataByServer<double>(_server, i, in[i]);
					break;

				case BOOL:
					consumeDataByServer<bool>(_server, i, in[i]);
					break;

				default:
					assert(false);
			}
		}

		in = _operationsRoot->pull();
		assert(in.size() != 0);
	}
}
